#pragma once
/*
    发布订阅功能:2种角色 [发布者 订阅者], 5个操作功能, 1个管理模块
    - 5个操作接口: 主题创建/ 删除/ 订阅/ 取消订阅/ 发布信息
    - 提供给dispatcher模块的消息分发处理模块
    - 1个管理: 主题名称与消息回调函数的映射关系. [订阅主题与主题消息发布是不同步的]
*/
#include "requestor.hpp"
#include <unordered_set>

namespace tmh_rpc
{
    namespace client
    {
        class TopicManager
        {
        public:
            using SubCallback = std::function<void(const string &, const string &)>;
            using ptr = std::shared_ptr<TopicManager>;

            TopicManager(const Requestor::ptr &requestor) : _requestor(requestor) {}
            // 主题创建
            bool create(const BaseConnection::ptr &conn, const string &key)
            {
                return commonRequest(conn, key, TopicOptype::TOPIC_CREATE);
            }
            // 主题删除
            bool remove(const BaseConnection::ptr &conn, const string &key)
            {
                return commonRequest(conn, key, TopicOptype::TOPIC_REMOVE);
            }
            // 主题订阅     因为订阅主题和接收到主题消息是不同步的, 因此在订阅主题的时候需要传入回调函数
            //[订阅的主题---主题消息接收处理回调函数]
            bool subscribe(const BaseConnection::ptr &conn, const string &key, const SubCallback &cb)
            {
                // 先创建主题
                addSubscribe(key, cb);
                bool ret = commonRequest(conn, key, TopicOptype::TOPIC_SUBSCRIBE);
                if (ret == false)
                {
                    // 主题订阅失败
                    delSubscribe(key); // 将刚刚创建添加的主题删除
                    return false;
                }
                return true;
            }
            // 取消订阅
            bool cancel(const BaseConnection::ptr &conn, const string &key)
            {
                delSubscribe(key);
                return commonRequest(conn, key, TopicOptype::TOPIC_CANCEL);
            }
            // 消息发布
            bool publish(const BaseConnection::ptr &conn, const string &key, const string &msg)
            {
                return commonRequest(conn, key, TopicOptype::TOPIC_PUBLISH, msg);
            }

            // 设置给dispatcher模块的回调处理函数: 对消息发布进行分发处理
            // 当收到消息发布请求后, 对这个请求进行分发管理
            void onPublish(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                // 1. 从消息中取出操作类型进行判断, 是否是消息请求
                auto type = msg->opType();
                if (type != TopicOptype::TOPIC_PUBLISH)
                {
                    ELOG("收到了错误类型的主题操作!");
                    return; 
                }
                // 2. 取出消息主题名称, 以及消息内容
                string topic_name = msg->topicKey();
                string topic_msg = msg->topicMsg();
                // 3. 通过主题名称, 查找对应的主题回调函数; 有则处理, 无就报错
                auto callback = getSubscribe(topic_name);
                if (!callback)
                {
                    ELOG("收到了 %s 主题消息, 但是该消息无主题处理回调!", topic_name.c_str());
                    return;
                }
                return callback(topic_name, topic_msg);
            }

        private:
            // ---------------- 针对map<主题名称, 回调函数> 的管理(增删查改) ------------------
            void addSubscribe(const string &key, const SubCallback &cb)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _topic_callbacks.insert({key, cb});
            }
            void delSubscribe(const string &key)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _topic_callbacks.erase(key);
            }
            const SubCallback getSubscribe(const string &key)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _topic_callbacks.find(key);
                if (it == _topic_callbacks.end())
                    return SubCallback();
                else
                    return it->second;
            }

            // ------------------------- 构建Topic请求, 向服务端发送 ---------------------------
            bool commonRequest(const BaseConnection::ptr &conn, const string &key,
                               TopicOptype type, const string &msg = "")
            {
                // 1. 构建请求
                auto topic_req = MessageFactory::create<TopicRequest>();
                topic_req->setId(UUID::uuid());
                topic_req->setMType(MType::REQ_TOPIC);
                topic_req->setOptype(type);
                topic_req->setTopicKey(key);
                if (type == TopicOptype::TOPIC_PUBLISH)
                {
                    // 只有当请求类型为消息发布, msg主体才会有内容
                    topic_req->setTopicMsg(msg);
                }
                // 2. 向服务端发送请求, 等待响应
                // [因为响应是异步的, 所以要借助requestor模块进行分发处理]
                BaseMessage::ptr rsp;
                bool ret = _requestor->send(conn, topic_req, rsp); // rsp是输入输出型参数
                if (ret == false)
                {
                    ELOG("主题操作请求失败!");
                    return false;
                }

                // 3. 判断请求处理是否成功
                auto topic_rsp = std::dynamic_pointer_cast<TopicResponse>(rsp);
                if (topic_rsp == nullptr)
                {
                    ELOG("主题操作响应, 向下类型转换失败!");
                    return false;
                }
                // 转换成功
                if (topic_rsp->rcode() != RCode::RCODE_OK)
                {
                    ELOG("主题操作请求出错: %s", errReason(topic_rsp->rcode()).c_str());
                    return false;
                }

                return true;
            }

        private:
            std::mutex _mutex;
            std::unordered_map<string, SubCallback> _topic_callbacks;

            // 对客户端的请求进行管理, 便于后续接收对应的响应[异步, 回调]
            Requestor::ptr _requestor;
        };
    }
}